import time

from confluent_kafka import Consumer, KafkaError, KafkaException, TopicPartition
import json
import sys

from lxml import etree

from cache.redis_util import getConnection
from dbse.mysql.db_entity_stock_kpi_day import StockKpiDay
# 配置日志
from selenium.common import TimeoutException

from mq.kafka.kafkaUtil import produceDfcfMsg
from utils.StringUtil import simpleReplace
from utils.DateUtil import getTodayStr
from dbse.mysql.db_mysql_insert import addStockKpi, addException

# logging.basicConfig(
#     format='%(asctime)s.%(msecs)03d [%(levelname)s] %(message)s',
#     datefmt='%Y-%m-%d %H:%M:%S',
#     level=logging.INFO
# )
# logger = logging.getLogger(__name__)

class DfcfStockKafkaConsumer:
    def handlePage(self,browser,stockCode,stockName,staDate,db,redis):
        try:
            print('==========handle page ==============')
            content = browser.page_source
            html = etree.fromstring(content,etree.HTMLParser())
            # 判断是否已退市
            error_state =  html.xpath('//div[@class="errorstate"]')
            if error_state:
                if error_state[0].text == '已退市':
                    redis.lpush('tplist',stockCode)                 ## 写入退市队列
                    print(f"==========股票{stockCode}:已退市")
                    return 1

            add_sql = 'INSERT INTO stock_dfcf_kpi_day (stock_code, stock_name, price_begin, price_end, price_max, price_min, price_increase, price_increase_rate, per, pbr,  sta_date, turnover_ratio, amount, total_capital, trunover, price_yesterday, vol_ratio) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'
            # 判断是否科创板
            if stockCode.startswith("688"):
                shoupan_result = html.xpath('//div[@class="zxj"]//span')
                zhangdie_result = html.xpath('//div[@class="zd"]/span')       # 两层span 嵌套 <span><span>xxx</span></span> 所以使用单 / 直接子元素
                result = html.xpath('//div[@class="brief_info_c"]//tbody//tr')          #
                row1_tds = result[0].getchildren()
                row2_tds = result[1].getchildren()
                row3_tds = result[2].getchildren()
                stockKpiDay = StockKpiDay(stockCode,stockName,
                                          row1_tds[5].find('span').find('span').text,
                                          shoupan_result[0].find('span').text,
                                          row1_tds[3].find('span').find('span').text,
                                          row2_tds[3].find('span').find('span').text,
                                          zhangdie_result[0].find('span').text,
                                          simpleReplace(zhangdie_result[1].find('span').text),
                                          row3_tds[7].find('span').find('span').text,
                                          row3_tds[9].find('span').find('span').text,
                                          # getTodayStr(), #原来的逻辑 取当前时间
                                          staDate,          # 新逻辑，取缓存中维护的当前交易日时间
                                          simpleReplace(row1_tds[7].find('span').find('span').text),
                                          simpleReplace(row3_tds[1].find('span').find('span').text),
                                          simpleReplace(row1_tds[9].find('span').find('span').text),
                                          simpleReplace(row2_tds[1].find('span').find('span').text),
                                          row2_tds[5].find('span').find('span').text,
                                          row2_tds[7].find('span').find('span').text
                                          )
                addStockKpi(db,add_sql,stockKpiDay)
                return 0
            else:
                shoupan_result = html.xpath('//div[@class="zxj"]//span')
                zhangdie_result = html.xpath('//div[@class="zd"]/span')       # 两层span 嵌套 <span><span>xxx</span></span> 所以使用单 / 直接子元素
                result = html.xpath('//div[@class="brief_info_c"]//tbody//tr')          #
                row1_tds = result[0].getchildren()
                row2_tds = result[1].getchildren()

                stockKpiDay = StockKpiDay(stockCode,stockName,
                                          row1_tds[1].find('span').find('span').text,
                                          shoupan_result[0].find('span').text,
                                          row1_tds[3].find('span').find('span').text,
                                          row2_tds[3].find('span').find('span').text,
                                          zhangdie_result[0].find('span').text,
                                          simpleReplace(zhangdie_result[1].find('span').text),
                                          row1_tds[11].find('span').find('span').text,
                                          row2_tds[11].find('span').find('span').text,
                                          # getTodayStr(),
                                          staDate,          # 新逻辑，取缓存中维护的当前交易日时间
                                          simpleReplace(row1_tds[7].find('span').find('span').text),
                                          simpleReplace(row2_tds[9].find('span').find('span').text),
                                          simpleReplace(row1_tds[13].find('span').find('span').text),
                                          simpleReplace(row1_tds[9].find('span').find('span').text),
                                          row2_tds[1].find('span').find('span').text,
                                          row2_tds[7].find('span').find('span').text
                                          )
                addStockKpi(db,add_sql,stockKpiDay)
                return 0
        except Exception as e:
            # print(f"异常信息: {e}")
            # print("详细堆栈信息:")
            # traceback.print_exc()  # 打印完整的堆栈跟踪
            print(f"======================{stockCode}发生异常======================")

            # 判断这条消息是否消费超过3次
            key = f'error:stock:kpi:{stockCode}'
            redis.incr(key)                                 # 记一次失败
            if(int(redis.get(key)) > 2):
                addException(db,stockCode)                  # 放弃这条数据，并记录在db
            else:
                produceDfcfMsg(stockCode,db)                # 失败的消息 重放队列
            time.sleep(1)
            return 1
            ## 重启任务
            # if(exceptionCount > 5):
            #     browser.quit()
            #     print("======================任务终止，等待重启======================")
            #     time.sleep(3)
            #     startWork(pJys,profile)

    def getPage(self,browser,stockCode,jys):
        try:
            url = f"https://quote.eastmoney.com/{jys}{stockCode}.html"
            browser.get(url)
        except TimeoutException:
            print("browser get overtime")               # browser设置了4S 超时短时间, 把结果交给后续处理，如果报错就记录信息和消息重放

    def getPage2(self):
        print('===========get page2===========')

    def __init__(self, config, topics,browser,db,batch,redis):
        """
        初始化Kafka消费者

        Args:
            config: 消费者配置字典
            topics: 要订阅的主题列表
        """
        self.consumer = Consumer(config)
        self.topics = topics
        self.running = True
        self.browser = browser
        self.db = db
        self.errorCount = 0
        self.batch = batch
        self.redis = redis

    def subscribe_to_topics(self):
        """订阅主题"""
        self.consumer.subscribe(self.topics)
        # if(partition != None):
        #     partition = TopicPartition('dfcf_stock_topic', partition)
        #     self.consumer.assign([partition])
        # logger.info(f"已订阅主题: {self.topics}")
        print(f"已订阅主题: {self.topics}")

    def consume_messages(self):
        """消费消息"""
        # try:
        while self.running:
            # 轮询消息，超时时间1秒
            time.sleep(1)
            msg = self.consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                # 错误处理
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # 分区末尾，正常情况
                    # logger.info(f"分区 {msg.partition()} 已到达末尾，偏移量: {msg.offset()}")
                    print(f"分区 {msg.partition()} 已到达末尾，偏移量: {msg.offset()}")
                else:
                    raise KafkaException(msg.error())
            else:
                # 成功接收到消息
                self.process_message(msg)
                try:
                    data = json.loads(msg.value().decode('utf-8'))
                except (json.JSONDecodeError, UnicodeDecodeError):
                    data = msg.value().decode('utf-8')

                stockCode = data["stockCode"]
                jys = data["jys"]
                stockName = data["stockName"]
                stockId = data["id"]
                staDate = data["staDate"]

                print(f'处理业务开始，stockId: {stockId}')
                # 抓股票页面 (有超时机制)
                self.getPage(self.browser,stockCode,jys)
                # 存数据
                errorTag = self.handlePage(self.browser,stockCode,stockName,staDate,self.db,self.redis)
                self.errorCount = self.errorCount + errorTag
                print(f'=================errorCount is: {self.errorCount}')
                if(self.errorCount > 5):
                    print('=============任务异常,重启任务============')
                    self.running = False
                    self.browser.quit()
                    self.close()
        return True #

        # except KeyboardInterrupt:
        #     logger.info("用户中断了消费过程")
        # except Exception as e:
        #     logger.error(f"消费过程中发生错误: {e}")
        # finally:
        #     # 关闭消费者
        #     self.close()

    def process_message(self, msg):
        """处理接收到的消息"""
        try:
            # 尝试解析JSON消息，如果不是JSON则直接使用原始值
            try:
                value = json.loads(msg.value().decode('utf-8'))
            except (json.JSONDecodeError, UnicodeDecodeError):
                value = msg.value().decode('utf-8')

            # 记录消息信息
            # logger.info(f"收到消息 [主题: {msg.topic()}, 分区: {msg.partition()}, 偏移量: {msg.offset()}]")
            # logger.info(f"消息键: {msg.key()}")
            # logger.info(f"消息值: {value}")
            print(f"收到消息 [主题: {msg.topic()}, 分区: {msg.partition()}, 偏移量: {msg.offset()}]")
            print(f"消息键: {msg.key()}")
            print(f"消息值: {value}")

            # 在这里添加你的业务逻辑
            # ...

            # 手动提交偏移量（如果配置了enable.auto.commit=false）
            # self.consumer.commit(asynchronous=False)

        except Exception as e:
            # logger.error(f"处理消息时发生错误: {e}")
            print(f"处理消息时发生错误: {e}")

    def close(self):
        """关闭消费者"""
        # logger.info("正在关闭消费者...")
        print("正在关闭消费者...")
        self.consumer.close()
        # logger.info("消费者已关闭")
        print("消费者已关闭")

    def quick_consume_messages(self):
        """消费消息"""
        try:
            while self.running:
                # 轮询消息，超时时间1秒
                msg = self.consumer.poll(1.0)
                if msg is None:
                    continue
                if msg.error():
                    # 错误处理
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        # 分区末尾，正常情况
                        # logger.info(f"分区 {msg.partition()} 已到达末尾，偏移量: {msg.offset()}")
                        print(f"分区 {msg.partition()} 已到达末尾，偏移量: {msg.offset()}")
                    else:
                        raise KafkaException(msg.error())
                else:
                    # 成功接收到消息
                    self.process_message(msg)

        except KeyboardInterrupt:
            # logger.info("用户中断了消费过程")
            print("用户中断了消费过程")
        except Exception as e:
            # logger.error(f"消费过程中发生错误: {e}")
            print(f"消费过程中发生错误: {e}")
        finally:
            # 关闭消费者
            self.close()

    def group_consume_messages(self):
        """消费消息"""
        try:
            while self.running:
                # 轮询消息，超时时间1秒
                time.sleep(1)
                msg = self.consumer.poll(1.0)
                if msg is None:
                    continue
                if msg.error():
                    # 错误处理
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        # 分区末尾，正常情况
                        # logger.info(f"分区 {msg.partition()} 已到达末尾，偏移量: {msg.offset()}")
                        print(f"分区 {msg.partition()} 已到达末尾，偏移量: {msg.offset()}")
                    else:
                        raise KafkaException(msg.error())
                else:
                    # 成功接收到消息
                    self.process_message(msg)

        except KeyboardInterrupt:
            # logger.info("用户中断了消费过程")
            print("用户中断了消费过程")
        except Exception as e:
            # logger.error(f"消费过程中发生错误: {e}")
            print(f"消费过程中发生错误: {e}")
        finally:
            # 关闭消费者
            self.close()

